package com.pisolution.scheduler.runtime.registry;

import com.google.common.base.Preconditions;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.DefaultACLProvider;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.regex.Pattern;

/**
 * 注册中心交互客户端.
 *
 * @author sora
 * @version 0.1.0
 */
@Component
public class ZookeeperClient implements InitializingBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperClient.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于配置客户端. */
    private final ZookeeperConfig config;

    /** 用于数据交互. */
    private CuratorFramework client;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    public ZookeeperClient(final ZookeeperConfig config) {
        this.config = config;
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    @Override
    public void afterPropertiesSet() {
        LOGGER.info("Try to initialize the registry center with the server list {}.", this.config.getQuorum());

        // 1. 校验命名空间是否合乎规则.
        Preconditions.checkArgument(
            Pattern.matches("^/[a-zA-Z]+/[a-zA-Z]+$", this.config.getRoot() + this.config.getNamespace()),
            "The namespace provided is invalid, please check if the namespace is not started with slash."
        );

        // 2. 配置客户端.
        this.client = CuratorFrameworkFactory
            .builder()
            .connectString(Preconditions.checkNotNull(this.config.getQuorum(), "Invalid zookeeper quorum."))
            .sessionTimeoutMs(this.config.getSessionTimeout())
            .connectionTimeoutMs(this.config.getConnectionTimeout())
            .retryPolicy(new ExponentialBackoffRetry(this.config.getBaseSleepTime(), this.config.getMaxRetries()))
            .namespace(this.config.getRoot().substring(1) + this.config.getNamespace())
            .aclProvider(new DefaultACLProvider())
            .build();

        // 3. 尝试连接服务端.
        this.client.start();

        // 4. 等待连接建立.
        try {
            this.client.blockUntilConnected();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        // 5. 监听客户端连接状态.
        this.client.getConnectionStateListenable().addListener((client, state) -> {
            switch (state) {
                case LOST:
                    LOGGER.error("The connection towards the registry center lost.");

                    break;
                case RECONNECTED:
                    LOGGER.info("The connection towards the registry center reestablished.");

                    break;
                case SUSPENDED:
                    LOGGER.warn("The connection towards the registry center suspended.");

                    break;
                default:
                    // 6. 剩下的状态暂时无需关注.

                    break;
            }
        });

        // 7. 建立节点缓存.
        this.nodeCache();
    }

    /**
     * 关闭客户端连接.
     */
    public void close() {
        CloseableUtils.closeQuietly(this.client);
    }

    // -------------------------------------------------------------------------------------------------
    //  属性访问
    // -------------------------------------------------------------------------------------------------

    public CuratorFramework getClient() {
        return this.client;
    }

    // -------------------------------------------------------------------------------------------------
    //  辅助方法 (生命周期)
    // -------------------------------------------------------------------------------------------------

    /**
     * 尝试将指定节点下的所有数据与子节点信息全部缓存到本地.
     */
    protected void nodeCache() {
    }

    /**
     * 监听指定节点的数据与结构变更情况.
     */
    protected void registerListener() {
    }

    // -------------------------------------------------------------------------------------------------
    //  辅助方法 (数据交互)
    // -------------------------------------------------------------------------------------------------

    /**
     * 校验给定节点是否存在于服务端.
     *
     * @param path 需要校验的节点路径.
     *
     * @return 如果存在就返回 {@code true} 否则返回 {@code false}.
     */
    public boolean exists(final String path) {
        try {
            return this.client.checkExists().forPath(path) != null;
        } catch (Exception e) {
            LOGGER.error("Could not check the existence of the given node path {}.", path, e);
        }

        return false;
    }

    /**
     * 将给定的数据存储于给定的永久性节点中 如果该节点不存在则创建该节点.
     *
     * @param path 数据存储的节点路径.
     * @param data 需要存储与给定节点的数据.
     */
    public void persist(final String path, final String data) {
        try {
            // 1. 如果节点存在则更新该节点数据.
            if (this.exists(path)) {
                this.update(path, data);

                return;
            }

            // 2. 创建该节点并存储数据.
            this.client
                .create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .forPath(path, data.getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            LOGGER.error("Could not store the data {} to the persistent path {}.", data, path, e);
        }
    }

    /**
     * 将给定的数据存储于给定的临时性节点中 如果该节点存在则移除该节点以便重新创建.
     *
     * @param path 数据存储的节点路径.
     * @param data 需要存储与给定节点的数据.
     */
    public void ephemeral(final String path, final String data) {
        try {
            // 1. 如果节点存在则移除该节点.
            if (this.exists(path)) {
                try {
                    this.client.delete().deletingChildrenIfNeeded().forPath(path);
                } catch (KeeperException.NoNodeException ignore) {
                    // 该节点已经被删除.
                }
            }

            // 2. 创建该节点并存储数据.
            this.client
                .create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL)
                .forPath(path, data.getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            LOGGER.error("Could not store the data {} to the ephemeral path {}.", data, path, e);
        }
    }

    /**
     * 以事务的方式更新给定节点中的数据.
     *
     * @param path 需要更新数据的节点路径.
     * @param data 需要存储与给定节点中的数据.
     */
    public void update(final String path, final String data) {
        try {
            this.client.transaction().forOperations(
                this.client.transactionOp().check().forPath(path),
                this.client.transactionOp().setData().forPath(path, data.getBytes(StandardCharsets.UTF_8))
            );
        } catch (Exception e) {
            LOGGER.error("Could not update the data of the path {}.", path, e);
        }
    }

    /**
     * 从服务端移除指定节点.
     *
     * @param path 需要移除的节点路径.
     */
    public void remove(final String path) {
        try {
            if (this.exists(path)) {
                this.client.delete().deletingChildrenIfNeeded().forPath(path);
            }
        } catch (KeeperException.NoNodeException ignore) {
            // 节点已经被删除 无需处理.
        } catch (Exception e) {
            LOGGER.error("Could not remove the path {}.", path, e);
        }
    }

    /**
     * 查询给定节点下的所有一级子节点路径.
     *
     * @param path 需要查询的节点路径.
     *
     * @return 该节点下的所有一级子节点路径.
     */
    public List<String> children(final String path) {
        try {
            return this.client.getChildren().forPath(path);
        } catch (InterruptedException e) {
            LOGGER.error("The action to iterate the path {} was interrupted.", path, e);

            throw new IllegalStateException(e);
        } catch (Exception e) {
            LOGGER.error("Could not iterate the path {}.", path, e);

            throw new RuntimeException(e);
        }
    }

    /**
     * 查询存储于给定节点中的数据.
     *
     * @param path 数据存储的节点路径.
     *
     * @return 该节点中存储的数据.
     */
    @Nullable
    public String get(final String path) {
        try {
            return new String(this.client.getData().forPath(path), StandardCharsets.UTF_8);
        } catch (Exception e) {
            LOGGER.error("Could not get the data stored in the path {}.", path, e);
        }

        return null;
    }
}
